package com.deep.flink;

import org.apache.flink.api.common.eventtime.SerializableTimestampAssigner;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

import java.time.Duration;

import static org.apache.flink.table.api.Expressions.$;

/**
 * @author sw
 * @create 2023-06-30 11:26
 */
public class Test9 {
    public static void main(String[] args) throws Exception {
//        StopWatch stopWatch = new StopWatch();
//        long start = stopWatch.getTime();

        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);

        SingleOutputStreamOperator<Event> eventStream = env.addSource(new ClickSource())
                .assignTimestampsAndWatermarks(WatermarkStrategy.<Event>forBoundedOutOfOrderness(Duration.ZERO)
                        .withTimestampAssigner(new SerializableTimestampAssigner<Event>() {
                            @Override
                            public long extractTimestamp(Event event, long l) {
                                return event.getTimestamp();
                            }
                        }));

        eventStream.print();
        //
        StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env);

        Table eventTable = tableEnv.fromDataStream(eventStream);
//        Table resultTable = eventTable.select($("username"), $("url")).where($("username").isEqual("Alice"));
        Table resultTable = eventTable.select($("username"), $("url"));

        tableEnv.createTemporaryView("clickTable", resultTable);

        //创建用于输出的表
        String createOutDDL = "CREATE TABLE `test5` (" +
                " username STRING," +
                " cnt BIGINT " +
                ") WITH (" +
                " 'connector' = 'filesystem'," +
                " 'path' = 'output'," +
                " 'format' = 'csv'" +
                ")";
        tableEnv.executeSql(createOutDDL);

        Table aggResult = tableEnv.sqlQuery("select username, ifnull(COUNT(url), 0) as cnt from clickTable group by username ");
        tableEnv.toChangelogStream(aggResult).print("agg");

        env.execute();

    }
}
